程序员练级攻略：异步I/O模型和Lock-Free编程-[2026重制版]

// 1. 创建epoll实例
int epfd = epoll_create1(0);
 
// 2. 注册事件
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET;  // 读事件 + 边缘触发
ev.data.fd = listen_fd;
epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev);
 
// 3. 等待事件（无限期）
#define MAX_EVENTS 1024
struct epoll_event events[MAX_EVENTS];
int nfds = epoll_wait(epfd, events, MAX_EVENTS, -1);
for (int i = 0; i < nfds; i++) {
    handle_event(events[i].data.fd, events[i].events);
}

// ET模式下必须设置非阻塞 + 循环读取
void handle_et_event(int fd) {
    char buf[4096];
    ssize_t nread;
 
    // 必须循环读取直到返回 EAGAIN
    while ((nread = read(fd, buf, sizeof(buf))) > 0) {
        process_data(buf, nread);
    }
 
    if (nread == -1 && errno != EAGAIN) {
        perror("read");
        close(fd);
    }
}
 
// 设置非阻塞I/O
int set_nonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}

#include <liburing.h>
 
#define QUEUE_DEPTH 256
 
int main() {
    struct io_uring ring;
    struct io_uring_sqe *sqe;
    struct io_uring_cqe *cqe;
 
    // 初始化io_uring实例
    io_uring_queue_init(QUEUE_DEPTH, &ring, 0);
 
    // 获取一个Submission Queue Entry
    sqe = io_uring_get_sqe(&ring);
 
    // 准备读操作（异步）
    char buf[4096];
    io_uring_prep_read(sqe, fd, buf, sizeof(buf), 0);
 
    // 提交请求
    io_uring_submit(&ring);
 
    // 等待完成
    io_uring_wait_cqe(&ring, &cqe);
 
    if (cqe->res < 0) {
        printf("Error: %s\n", strerror(-cqe->res));
    } else {
        printf("Read %d bytes\n", cqe->res);
    }
 
    // 标记完成项已处理
    io_uring_cqe_seen(&ring, cqe);
    io_uring_queue_exit(&ring);
    return 0;
}

# Python asyncio风格的Reactor实现
import selectors
import socket
 
class Reactor:
    def __init__(self):
        self.selector = selectors.DefaultSelector()
        self.handlers = {}
 
    def register(self, fd, event_mask, callback):
        """注册文件描述符和对应的事件处理器"""
        self.selector.register(fd, event_mask, callback)
 
    def unregister(self, fd):
        self.selector.unregister(fd)
 
    def run(self):
        """主事件循环"""
        while True:
            events = self.selector.select(timeout=1)
            for key, mask in events:
                callback = key.data
                callback(key.fileobj, mask)
 
 
# 使用示例
def accept_connection(sock, mask):
    conn, addr = sock.accept()
    print(f"Accepted connection from {addr}")
    conn.setblocking(False)
    reactor.register(conn, selectors.EVENT_READ, handle_read)
 
def handle_read(conn, mask):
    data = conn.recv(4096)
    if data:
        print(f"Received: {data.decode()}")
        # 切换为写事件
        reactor.modify(conn, selectors.EVENT_WRITE, handle_write)
    else:
        reactor.unregister(conn)
        conn.close()
 
def handle_write(conn, mask):
    conn.send(b"HTTP/1.1 200 OK\r\nContent-Length: 13\r\n\r\nHello, World!")
    reactor.unregister(conn)
    conn.close()
 
 
reactor = Reactor()
server_sock = socket.socket()
server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_sock.bind(('0.0.0.0', 8888))
server_sock.listen(128)
server_sock.setblocking(False)
 
reactor.register(server_sock, selectors.EVENT_READ, accept_connection)
reactor.run()

// C++11 std::atomic CAS操作示例
#include <atomic>
#include <iostream>
 
std::atomic<int> counter(0);
 
void increment() {
    int expected, desired;
    do {
        expected = counter.load();     // 读取当前值
        desired = expected + 1;         // 计算新值
        // 如果counter == expected，则设置为desired
        // 否则重新尝试（自旋）
    } while (!counter.compare_exchange_weak(expected, desired));
}
 
// 使用C++20的atomic_ref实现更细粒度的无锁操作
template<typename T>
class LockFreeStack {
private:
    struct Node {
        T data;
        Node* next;
    };
    std::atomic<Node*> head;
 
public:
    void push(T value) {
        Node* new_node = new Node{value, nullptr};
        do {
            new_node->next = head.load();
        } while (!head.compare_exchange_weak(new_node->next, new_node));
    }
 
    bool pop(T& result) {
        Node* old_head;
        do {
            old_head = head.load();
            if (!old_head) return false;
        } while (!head.compare_exchange_weak(old_head, old_head->next));
        result = old_head->data;
        delete old_head;
        return true;
    }
};

std::atomic<int> data_ready{0};
int shared_data = 0;
 
// 生产者线程
void producer() {
    shared_data = 42;  // 写入数据
 
    // release语义：确保shared_data的写入对其他线程可见
    data_ready.store(1, std::memory_order_release);
}
 
// 消费者线程
void consumer() {
    // acquire语义：确保读到data_ready==1时，能看到之前的所有写入
    while (!data_ready.load(std::memory_order_acquire)) {
        std::this_thread::yield();  // 或 busy-wait
    }
 
    // 此时一定能看到 shared_data == 42
    std::cout << "Data: " << shared_data << std::endl;
}